/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package com.hp.bon.sgw.scsp;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.mina.core.RuntimeIoException;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.AbstractIoConnector;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.ProtocolDecoderException;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hp.bon.sgw.scsp.SCSPConstants.ResultCode;
import com.hp.bon.sgw.util.Comutils;

/**
 * {@link IoHandler} for SumUp server.
 * 
 * @author <a href="http://mina.apache.org">Apache MINA Project</a>
 */
public class SessionHandler extends IoHandlerAdapter {

	protected final static Logger LOGGER = LoggerFactory
			.getLogger(SessionHandler.class);

	/**
	 * session => request message map
	 * <p>
	 * Key :session
	 * <p>
	 * Value :request message map,Map
	 * <hr/>
	 * request message map:mid => message
	 * <p>
	 * Key :MID
	 * <p>
	 * Value :request message when sent request
	 */
	public static final ConcurrentHashMap<IoSession, ConcurrentHashMap<String, SCSPMessage>> session2MsgMap = new ConcurrentHashMap<IoSession, ConcurrentHashMap<String, SCSPMessage>>();

	public enum State {
		connecting, connected_in, connected_out, tls, ready, closed
	}

	final ConcurrentHashMap<String, Peer> _persistentPeers = new ConcurrentHashMap<String, Peer>();
	protected final ConcurrentHashMap<String, IoSession> _activeSessions = new ConcurrentHashMap<String, IoSession>();

	/**
	 * add a Persistent peer
	 * 
	 * @param peer
	 */

	public void addPersistentPeer(Peer peer) {
		synchronized (_persistentPeers) {
			_persistentPeers.put(peer.getHostID(), peer);
			_persistentPeers.notifyAll();
		}
		LOGGER.info("Add a Persistent Peer: " + peer);
	}

	public void addAllPersistentPeer(ConcurrentHashMap<String, Peer> newMap) {
		synchronized (_persistentPeers) {
			for (String hostName : _persistentPeers.keySet()) {
				if (!newMap.containsKey(hostName)) {
					LOGGER.info("Remove a Persistent Peer: "
							+ _persistentPeers.remove(hostName));
				}
			}
			for (String hostName : newMap.keySet()) {
				Peer peer = _persistentPeers.get(hostName);
				if (peer == null) {
					peer = newMap.get(hostName);
					_persistentPeers.put(hostName, peer);
					LOGGER.info("Add a Persistent Peer: " + peer);
				}
			}
			_persistentPeers.notifyAll();
		}
	}

	/**
	 * Remove a Persistent peer
	 * 
	 * @param peer
	 */
	public void removePersistentPeer(Peer peer) {
		synchronized (_persistentPeers) {
			_persistentPeers.remove(peer.getHostID());
			_persistentPeers.notifyAll();
		}
		LOGGER.info("Remove a Persistent Peer: " + peer.toString());
	}

	/**
	 * Reconnect thread for persistent connection.
	 */
	private class ReconnectThread extends Thread {

		private AbstractIoConnector connector;

		public void run() {
			while (true) {
				for (String peerName : _persistentPeers.keySet()) {
					Peer peer = _persistentPeers.get(peerName);
					runPeer(peer);
				}
				synchronized (_persistentPeers) {
					try {
						_persistentPeers.wait(reconnectRetryTime * 1000);
					} catch (InterruptedException e) {
					}
				}
			}
		}

		public void runPeer(Peer peer) {
			if (_activeSessions.containsKey(peer.getHostID())) {
				return;
			}
			try {
				ConnectFuture future = connector.connect(new InetSocketAddress(
						peer.getHostAddress(), peer.getPort()));
				future.awaitUninterruptibly();
				IoSession session = future.getSession();
				session.setAttribute(SCSPConstants.ATTR_PEER_STATE,
						SessionHandler.State.connected_out);
				session.setAttribute(SCSPConstants.ATTR_HOST_ID,
						peer.getHostID());
				Element rootDWR = DocumentHelper.createElement(MSG_TYPE_CLR);
				String attrMID = SCSPUtils.makeSessionID(hostID);
				rootDWR.addAttribute(MSG_FIEDD_MID, attrMID);
				rootDWR.addAttribute(MSG_FIEDD_SN, getHostID());
				rootDWR.addAttribute(MSG_FIEDD_UID, getHostID());
				rootDWR.addAttribute(MSG_FIEDD_PWD, peer.getAuthCode());
				sendMsg(session, rootDWR, false, null);
			} catch (RuntimeIoException e) {
				if (LOGGER.isInfoEnabled()) {
					LOGGER.info("Failed to connect " + peer);
				}

			}

		}

		public ReconnectThread(NioSocketConnector connector) {
			this.connector = connector;
		}
	}

	/**
	 * Retransmitting thread for persistent connection.
	 */
	private class RetransmittingTask extends Thread {

		public void run() {
			while (true) {
				long now = System.currentTimeMillis();
				for (IoSession session : session2MsgMap.keySet()) {
					if (session.isClosing()) {
						LOGGER.info("session is closing,remove from SessionMap:"
								+ session.getRemoteAddress());
						session2MsgMap.remove(session);
					} else {
						runSession(session, now);
					}
				}
				try {
					Thread.sleep(3 * 1000);
				} catch (InterruptedException e) {

				}
			}
		}

		public void runSession(IoSession session, long now) {
			ConcurrentHashMap<String, SCSPMessage> msgMap = session2MsgMap
					.get(session);
			ArrayList<String> msgIdList = new ArrayList<String>();
			for (String sessionId : msgMap.keySet()) {
				SCSPMessage msg = msgMap.get(sessionId);
				if (msg == null)
					continue;
				if (now - msg._st1 > MsgTimeOut) {
					if (MsgResendTimes == 0
							|| msg.getResendTimes() >= MsgResendTimes) {
						msgIdList.add(sessionId);
					} else {
						msg.setResendTimes(msg.getResendTimes() + 1);
						LOGGER.info("retransmit msg to "
								+ session
										.getAttribute(SCSPConstants.ATTR_HOST_ID)
								+ "|"
								+ session
										.getAttribute(SCSPConstants.ATTR_Session_Info)
								+ msg.getFormatBody());
						session.write(msg);
					}
				}
			}
			for (String sessionId : msgIdList) {
				SCSPMessage msg = msgMap.remove(sessionId);
				LOGGER.info("session timeout,remove from SessionMap:"
						+ msg.getBodyStr());
			}
		}

		public RetransmittingTask() {
			super("RetransmittingTask-" + hostID);
			// setPriority(Thread.MIN_PRIORITY);
		}
	}

	/**
	 * Do election by host id
	 * 
	 * @param hostID
	 * @return True if I win
	 */
	public int doElection(String hostID) {
		if (hostID.equals(this.hostID)) {
			LOGGER.warn("Local name is the same as remote name:" + hostID);
			return ResultCode.ERR_Remote_And_Local_Same_Name;
		}
		IoSession session = _activeSessions.get(hostID);
		if (session != null) {
			LOGGER.warn("Found a exist active peer "
					+ session.getRemoteAddress()
					+ " while received a CLR from " + hostID);
			return ResultCode.ERR_Dup_Connection;
		}

		return ResultCode.SUCCESS;
	}

	public void startAsServer() {
		NioSocketAcceptor acceptor = new NioSocketAcceptor();
		acceptor.getFilterChain().addLast("codec",
				new ProtocolCodecFilter(new SCSPProtocolCodecFactory()));
		acceptor.getFilterChain().addLast("logger", new LoggingFilter());
		acceptor.getFilterChain().addLast("threadPool",
				new ExecutorFilter(useThreadPool(maximumPoolSize)));

		acceptor.setHandler(this);
		InetSocketAddress address = new InetSocketAddress(this.getPort());
		try {
			acceptor.bind(address);
		} catch (IOException e) {
			LOGGER.error("stopped because binding failed:" + address);
			System.exit(-1);
		}
		LOGGER.info("Listening on port " + this.getPort());
		startAsClient();

	}

	public void startAsClient() {

		if (LOGGER.isDebugEnabled())
			LOGGER.debug("startAsClient...");
		NioSocketConnector connector = new NioSocketConnector();
		connector.getFilterChain().addLast("codec",
				new ProtocolCodecFilter(new SCSPProtocolCodecFactory()));
		connector.getFilterChain().addLast("logger", new LoggingFilter());
		connector.getFilterChain().addLast("threadPool",
				new ExecutorFilter(useThreadPool(maximumPoolSize)));
		connector.setHandler(this);

		LOGGER.info("Begin Starting ReconnectThread ..");
		ReconnectThread reconnectThread = new ReconnectThread(connector);
		reconnectThread.setDaemon(true);
		reconnectThread.start();
		LOGGER.info("Start ReconnectThread OK");

		LOGGER.info("Begin Starting RetransmittingTask ..");
		RetransmittingTask retransmittingTask = new RetransmittingTask();
		retransmittingTask.setDaemon(true);
		retransmittingTask.start();
		LOGGER.info("Start RetransmittingTask OK");
	}

	private ThreadPoolExecutor _threadPool;

	public ThreadPoolExecutor useThreadPool(int maximumPoolSize) {
		if (_threadPool == null) {
//			_threadPool = Executors.newCachedThreadPool();
			 _threadPool = new ThreadPoolExecutor(0, maximumPoolSize,
			 60L, TimeUnit.SECONDS,
			 new SynchronousQueue<Runnable>());
		}
		return _threadPool;
	}

	@Override
	public void sessionOpened(IoSession session) {

		String sessionInfo = session.getLocalAddress() + "|"
				+ session.getRemoteAddress();
		LOGGER.info("socket opened:" + sessionInfo);
		session.setAttribute(SCSPConstants.ATTR_Session_Info, sessionInfo);
		session.getConfig().setIdleTime(IdleStatus.READER_IDLE, watchdogTimout);
	}

	public void sessionClosed(IoSession session) throws Exception {
		String hostID = (String) session
				.getAttribute(SCSPConstants.ATTR_HOST_ID);
		if (hostID != null) {
			IoSession sucessSession = _activeSessions.remove(hostID);
			if (sucessSession != null) {
				onConnectionClosed(hostID, sucessSession);
			}
		}
		LOGGER.warn("socket[" + hostID + "] closed from "
				+ session.getLocalAddress() + " to "
				+ session.getRemoteAddress());
	}

	@Override
	public void messageReceived(IoSession session, Object message) {

		SCSPMessage recvMsg = (SCSPMessage) message;
		String strBody = recvMsg.getBodyStr();
		State state = (State) session.getAttribute(
				SCSPConstants.ATTR_PEER_STATE,
				SessionHandler.State.connected_in);
		try {
			Element root = Comutils.parseText(strBody).getRootElement();
			recvMsg.setElement(root);
			if (LOGGER.isInfoEnabled()) {
				String formatBody = SCSPUtils.formatMessage(recvMsg);
				recvMsg.setFormatBody(formatBody);
				LOGGER.info("recv msg from "
						+ session.getAttribute(SCSPConstants.ATTR_HOST_ID)
						+ "|"
						+ session.getAttribute(SCSPConstants.ATTR_Session_Info)
						+ recvMsg.getFormatBody());
			}
			String msgType = root.getName();
			if (state == SessionHandler.State.connected_in) {
				if (msgType.equals(MSG_TYPE_CLR)) {
					handleCLR(session, root);
					return;
				} else {
					closeSocket(session,
							"recv msg not CLR after connnection established",
							ResultCode.ERR_NO_EXPECTED_MSG);
				}
			} else if (state == SessionHandler.State.connected_out) {
				if (msgType.equals(MSG_TYPE_CLA)) {
					handleCLA(session, root);
					return;
				} else {
					closeSocket(session, "recv msg not CLA after send CLR",
							ResultCode.ERR_NO_EXPECTED_MSG);
				}
			}
			if (msgType.equals(MSG_TYPE_SCR)) {
				handleSCR(session, root);

			} else if (msgType.equals(MSG_TYPE_SCA)) {
				String attrMID = SCSPUtils
						.parseField(root, MSG_FIEDD_MID, true);
				ConcurrentHashMap<String, SCSPMessage> mapName2Request = session2MsgMap
						.get(session);
				SCSPMessage requestMsg = null;
				if (mapName2Request != null) {
					requestMsg = mapName2Request.remove(attrMID);
				}
				if (requestMsg != null) {
					handleSCA(session, root, requestMsg);
				} else {
					LOGGER.info("Not found request for answer:" + root.asXML());
				}

			} else if (msgType.equals(MSG_TYPE_DWR)) {
				handleDWR(session, root);
			} else if (msgType.equals(MSG_TYPE_DWA)) {
				LOGGER.debug("Receive DWA from "
						+ session.getAttribute(SCSPConstants.ATTR_HOST_ID));
			} else if (msgType.equals(MSG_TYPE_DPR)) {
				handleDPR(session, root);
			} else {
				LOGGER.warn("recv unknown msg type:" + strBody);
			}
		} catch (Exception e) {
			String errInfo = "recv msg is not XML format:";
			LOGGER.warn(errInfo + strBody);
			this.closeSocket(session, errInfo, ResultCode.ERR_XML_FORMAT_FAILED);
		}
	}

	/**
	 * @param session
	 * @param root
	 * @throws FieldNotExistException
	 */
	private void handleDWR(IoSession session, Element root)
			throws FieldNotExistException {
		if (_debugDisableReplyWatchdog)
			return;
		String attrMID = SCSPUtils.parseField(root, MSG_FIEDD_MID, true);

		Element rootDWA = DocumentHelper.createElement(MSG_TYPE_DWA);
		rootDWA.addAttribute(MSG_FIEDD_MID, attrMID);
		rootDWA.addAttribute(MSG_FIEDD_RTT, Comutils.formatTime(
				System.currentTimeMillis(), "yyyyMMddhhmmssSSS"));
		rootDWA.addAttribute(MSG_FIEDD_RC, "0");
		this.sendMsg(session, rootDWA, false, null);
	}

	/**
	 * @param session
	 * @param root
	 * @throws FieldNotExistException
	 */
	private void handleDPR(IoSession session, Element root)
			throws FieldNotExistException {
		String attrMID = SCSPUtils.parseField(root, MSG_FIEDD_MID, true);

		Element rootDPA = DocumentHelper.createElement(MSG_TYPE_DPA);
		rootDPA.addAttribute(MSG_FIEDD_MID, attrMID);
		rootDPA.addAttribute(MSG_FIEDD_RTT, Comutils.formatTime(
				System.currentTimeMillis(), "yyyyMMddhhmmssSSS"));
		rootDPA.addAttribute(MSG_FIEDD_RC, "0");
		this.sendMsg(session, rootDPA);
		session.close(false);
	}

	/**
	 * 能力交换函数，在建立连接是做接入鉴权
	 * 
	 * @param session
	 * @param request
	 * @param answer
	 * @return
	 */
	protected int onConnectionCapbilityExchanging(IoSession session,
			String userName, String password) {
		return ResultCode.SUCCESS;
	}

	/**
	 * 连接断开时调用此函数，用于更新链路状态 连接建立完毕后调用此函数，用于更新链路状态
	 * 
	 * @param session
	 */
	protected void closeSocket(IoSession session, String errInfo, int errCode) {
		Element rootDPR = DocumentHelper.createElement(MSG_TYPE_DPR);
		String attrMID = SCSPUtils.makeSessionID(this.hostID);
		rootDPR.addAttribute(MSG_FIEDD_MID, attrMID);
		rootDPR.addAttribute(MSG_FIEDD_RTT, Comutils.formatTime(
				System.currentTimeMillis(), "yyyyMMddhhmmssSSS"));
		rootDPR.addAttribute(MSG_FIEDD_RC, String.valueOf(errCode));
		SCSPUtils.addElement(rootDPR, MSG_NODE_DS, errInfo);
		this.sendMsg(session, rootDPR);
		LOGGER.warn("closeing socket from " + session.getLocalAddress()
				+ " to " + session.getRemoteAddress() + ",errInfo:" + errInfo);
		session.close(false);
	}

	/**
	 * 连接关闭时调用此函数,该连接已经成功建立,有子类来更新状态
	 * 
	 * @param hostID
	 * @param session
	 */
	public void onConnectionClosed(String hostID, IoSession session) {

	}

	/**
	 * 连接建立完毕后调用此函数，用于更新链路状态
	 * 
	 * @param hostID
	 * @param session
	 */
	public void onConnectionReady(String hostID, IoSession session) {
		session.setAttribute(SCSPConstants.ATTR_PEER_STATE,
				SessionHandler.State.connecting);
		session.setAttribute(SCSPConstants.ATTR_HOST_ID, hostID);
		_activeSessions.put(hostID, session);
		LOGGER.info("connect susscess with：" + hostID);
	}

	/**
	 * 处理SCA， 根据SCA的sessionID从发送Map中查找SCR，如果找到则返回SCR的Element，否则返回null
	 */
	public void handleSCA(IoSession session, Element answerRoot, SCSPMessage req) {
	}

	public static ConcurrentHashMap<IoSession, ConcurrentHashMap<String, SCSPMessage>> getSession2MsgMap() {
		return session2MsgMap;
	}

	/**
	 * 基类实现方法，收到SCR后生成对应的SCA返回
	 * 
	 * @param session
	 * @param root
	 */
	public void handleSCR(IoSession session, Element root) {

		try {
			String attrMID = SCSPUtils.parseField(root, MSG_FIEDD_MID, true);
			String attrSCID = SCSPUtils.parseField(root, MSG_FIEDD_SCID, true);
			Element rootSCA = DocumentHelper.createElement(MSG_TYPE_SCA);
			rootSCA.addAttribute(MSG_FIEDD_MID, attrMID);
			rootSCA.addAttribute(MSG_FIEDD_SCID, attrSCID);
			rootSCA.addAttribute(MSG_FIEDD_RC, "0");
			SCSPUtils.addElement(rootSCA, MSG_NODE_DS,
					"return by CLASS SessionHandler");
			Element serviceInformation = DocumentHelper
					.createElement("Service_Information");
			Element serviceResultCode = DocumentHelper
					.createElement("Service_Result_Code");
			serviceResultCode.setText("0");
			serviceInformation.add(serviceResultCode);
			// DATA Convert begin
			// SCSPUtils.addElement(rootSCA, MSG_NODE_DATA,
			// serviceInformation.asXML());
			// DATA no Convert begin
			SCSPUtils.addElement(rootSCA, MSG_NODE_DATA, serviceInformation);
			this.sendMsg(session, rootSCA, false, new Object());
		} catch (FieldNotExistException e) {
			e.printStackTrace();
		}
	}

	// sendRequestSyn同步方式发送SCR，收到SCA后，唤醒请求线程，返回SCA
	// public void handleSCA(IoSession session, Element answerRoot,SCSPMessage
	// req) {
	//
	// Element requestRoot = req.getElement();
	// if (request2Answer.containsKey(requestRoot)) {
	// request2Answer.put(requestRoot, answerRoot);
	// synchronized (requestRoot) {
	// requestRoot.notifyAll();
	// }
	// }
	// }
	// ConcurrentHashMap<Element, Element> request2Answer = new
	// ConcurrentHashMap<Element, Element>();
	//
	// /**
	// * 同步方式发送SCR，等待SCA返回
	// *
	// * @param hostId
	// * @param rootSCR
	// * @return
	// */
	// public Element sendRequestSyn(String hostId, Element rootSCR) {
	// request2Answer.put(rootSCR, rootSCR);
	// this.sendRequest(hostId, rootSCR,null);
	// synchronized (rootSCR) {// 等待调用结果完成
	// try {
	// rootSCR.wait(30 * 1000);
	// } catch (InterruptedException e) {
	// LOGGER.warn(" executeRemoteCall InterruptedException ", e);
	// }
	// }
	// return request2Answer.remove(rootSCR);
	// }
	public int sendRequest(String hostID, Element root, Object stateObject) {
		IoSession session = this._activeSessions.get(hostID);
		if (session == null) {
			LOGGER.info("Not found socket for " + hostID);
			return ResultCode.ERR_SESSION_NOT_EXIST;
		}
		return sendMsg(session, root, true, stateObject);
	}

	public int sendMsg(IoSession session, Element root) {
		return this.sendMsg(session, root, false, null);
	}

	public int sendMsg(IoSession session, Element root, boolean isBuzRequest,
			Object stateObject) {
		String attrMID = root.attributeValue(SessionHandler.MSG_FIEDD_MID);
		if (attrMID == null) {
			LOGGER.info("send failed because MID missed: " + root.asXML());
			return ResultCode.ERR_MID_MISSED;
		}

		if (session.isClosing()) {
			LOGGER.info("socket["
					+ session.getAttribute(SCSPConstants.ATTR_Session_Info)
					+ "] is closing,send message failed: " + root.asXML());
			return ResultCode.ERR_SESSION_CLOSED;
		}
		SCSPMessage msg = new SCSPMessage(root);
		if (isBuzRequest) {
			msg.setStateObject(stateObject);
			// 设置发送时间
			msg._st1 = System.currentTimeMillis();
			ConcurrentHashMap<String, SCSPMessage> mapName2Request = session2MsgMap
					.get(session);
			if (mapName2Request == null) {
				mapName2Request = new ConcurrentHashMap<String, SCSPMessage>();
				session2MsgMap.put(session, mapName2Request);
			}
			mapName2Request.put(attrMID, msg);
		}
		if (LOGGER.isInfoEnabled()) {
			String formatBody = null;
			try {
				formatBody = SCSPUtils.formatMessage(msg);
			} catch (Exception e) {
				LOGGER.warn("msg format failed :" + root.toString());
				return ResultCode.ERR_XML_FORMAT_FAILED;
			}
			msg.setFormatBody(formatBody);
			LOGGER.info("send msg to "
					+ session.getAttribute(SCSPConstants.ATTR_HOST_ID) + "|"
					+ session.getAttribute(SCSPConstants.ATTR_Session_Info)
					+ msg.getFormatBody());
		}
		session.write(msg);
		return ResultCode.SUCCESS;
	}

	/**
	 * @param session
	 * @param root
	 * @throws FieldNotExistException
	 */
	protected void handleCLR(IoSession session, Element root) {
		String attrMID;
		try {
			attrMID = SCSPUtils.parseField(root, MSG_FIEDD_MID, true);
			String attrUID = SCSPUtils.parseField(root, MSG_FIEDD_UID, true);
			Element rootCLA = DocumentHelper.createElement(MSG_TYPE_CLA);
			rootCLA.addAttribute(MSG_FIEDD_MID, attrMID);
			rootCLA.addAttribute(MSG_FIEDD_RC, "0");
			Element eleDS = DocumentHelper.createElement(MSG_NODE_DS);
			eleDS.setText("Return by the base class");
			rootCLA.add(eleDS);
			onConnectionReady(attrUID, session);
			this.sendMsg(session, rootCLA);
		} catch (FieldNotExistException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	private void handleCLA(IoSession session, Element root) {
		String strResult = root.attributeValue(MSG_FIEDD_RC, "-1");
		String hostID = (String) session
				.getAttribute(SCSPConstants.ATTR_HOST_ID);
		int result = Integer.parseInt(strResult);
		if (result == ResultCode.SUCCESS) {
			onConnectionReady(hostID, session);
		} else {
			closeSocket(session, "CLA result is " + result,
					ResultCode.ERR_SESSION_CLOSED);
		}
	}

	@Override
	public void sessionIdle(IoSession session, IdleStatus status) {

		State state = (State) session.getAttribute(
				SCSPConstants.ATTR_PEER_STATE,
				SessionHandler.State.connected_in);
		if (state == SessionHandler.State.connected_in) {
			closeSocket(session, "not recv CLR after connected",
					ResultCode.ERR_NO_EXPECTED_MSG);
		} else if (state == SessionHandler.State.connected_out) {
			closeSocket(session, "not recv CLA after connected",
					ResultCode.ERR_NO_EXPECTED_MSG);
		}
		int idleCount = session.getIdleCount(IdleStatus.READER_IDLE);
		if (idleCount >= watchdogMax) {
			closeSocket(session, "not recv msg after send DWR " + watchdogMax
					+ " times", ResultCode.ERR_DWR_EXCEPTION);
			return;
		}
		// For debug
		if (_debugDisableReplyWatchdog)
			return;
		String attrMID = SCSPUtils.makeSessionID(this.hostID);
		Element rootDWR = DocumentHelper.createElement(MSG_TYPE_DWR);
		rootDWR.addAttribute(MSG_FIEDD_MID, attrMID);
		this.sendMsg(session, rootDWR, false, null);
	}

	@Override
	public void exceptionCaught(IoSession session, Throwable cause) {

		if (cause != null && cause instanceof ProtocolDecoderException) {
			this.closeSocket(session, "SCSP version exception!",
					ResultCode.ERR_SCSP_VERION_EXCEPTION);
		} else {
			session.close(false);
		}
		LOGGER.warn("exceptionCaught from " + session.getLocalAddress()
				+ " to " + session.getRemoteAddress() + ",errInfo:"
				+ cause.getMessage());

	}

	public static final String MSG_TYPE_CLR = "CLR";
	public static final String MSG_TYPE_CLA = "CLA";
	public static final String MSG_TYPE_DWR = "DWR";
	public static final String MSG_TYPE_DWA = "DWA";
	public static final String MSG_TYPE_SCR = "SCR";
	public static final String MSG_TYPE_SCA = "SCA";
	public static final String MSG_TYPE_DPR = "DPR";
	public static final String MSG_TYPE_DPA = "DPA";

	public static final String MSG_FIEDD_MID = "MID";
	public static final String MSG_FIEDD_UID = "UID";
	public static final String MSG_FIEDD_PWD = "PWD";
	public static final String MSG_FIEDD_SN = "SN";
	public static final String MSG_FIEDD_REPTAG = "REPTAG";
	public static final String MSG_FIEDD_SCID = "SCID";
	public static final String MSG_FIEDD_RTT = "RTT";
	public static final String MSG_FIEDD_RC = "RC";
	public static final String MSG_NODE_SIG = "SIG";
	public static final String MSG_NODE_DS = "DS";
	public static final String MSG_NODE_DATA = "DATA";

	protected int port = 8080;
	protected String hostID = "SCSP@123.com";
	protected String hostAddress = "127.0.0.1";
	protected String userName = "";
	protected String password = "";
	private int watchdogMax = 3;
	private int watchdogTimout = 120;
	private int reconnectRetryTime = 30;
	private int maximumPoolSize = 10;
	private int MsgTimeOut = 60 * 1000;// 单位毫秒
	private int MsgResendTimes = 0;// 单位秒

	public void setHostConf(String hostID, String hostAddress, int port,
			String userName, String password) {
		this.hostID = hostID;
		if (hostAddress.equals("127.0.0.1") || hostAddress.equals("localhost")) {
			this.hostAddress = "0.0.0.0";
		} else {
			this.hostAddress = hostAddress;
		}

		this.port = port;
		this.userName = userName;
		this.password = password;
	}

	public void setConnectionTimeout(int maxWatchdog, int watchdogTimeout) {
		this.watchdogMax = maxWatchdog;
		this.watchdogTimout = watchdogTimeout;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public String getHostID() {
		return hostID;
	}

	public void setHostID(String hostID) {
		this.hostID = hostID;
	}

	public int getReconnectRetryTime() {
		return reconnectRetryTime;
	}

	public void setReconnectRetryTime(int reconnectRetryTime) {
		this.reconnectRetryTime = reconnectRetryTime;
	}

	public void setMaximumPoolSize(int maximumPoolSize) {
		this.maximumPoolSize = maximumPoolSize;
	}

	public boolean isSessionReady(String hostId) {
		return _activeSessions.containsKey(hostId);
	}

	public ConcurrentHashMap<String, IoSession> getActiveSession() {
		return this._activeSessions;
	}

	boolean _debugDisableReplyWatchdog = false;// for DEBUG

	public void setDebugDisbaleReplyWacthdog(boolean disable) {
		_debugDisableReplyWatchdog = disable;
	}

	public String toString() {
		StringWriter sw = new StringWriter();
		PrintWriter pw = new PrintWriter(sw);
		pw.printf("========= %s Firmware %d=============\n",
				"HP SGW SCSP Simulator V0.1(LCZ)", 001);
		pw.printf("SCSP ID %s on %s:%d\n", this.hostID, this.hostAddress,
				this.port);
		Runtime runtime = Runtime.getRuntime();
		pw.printf(
				"JVM memory free %dK , total %dK , max %dK , Active Thread %d.\n",
				runtime.freeMemory() / 1024, runtime.totalMemory() / 1024,
				runtime.maxMemory() / 1024, Thread.activeCount());
		// pw.printf("Thread pool active %d thread,task queue %d,remaining %d,largest queue size %d.\n",
		// this._threadPool.
		// , getThreadPoolQueueSize(), getThreadPoolQueueRemaining(),
		// getThreadPoolLargestQueueSize());
		// NodeStatUtils stat = NodeStatUtils.getInstance();
		// if (stat.isStartPeakSample())
		// {
		// pw.printf("Node Peak receive %d in %dms(%d Msg/s),send %d in %dms(%d Msg/s)\n",
		// stat._peakMaxReceived, stat._peakTimespan, stat._peakMaxReceived *
		// 1000 / stat._peakTimespan,
		// stat._peakMaxSent, stat._peakTimespan, stat._peakMaxSent * 1000 /
		// stat._peakTimespan);
		// }
		pw.close();
		return sw.toString();
	}

	public int getMsgTimeOut() {
		return MsgTimeOut;
	}

	public void setMsgTimeOut(int msgTimeOut) {
		MsgTimeOut = msgTimeOut;
	}

	public int getMsgResendTimes() {
		return MsgResendTimes;
	}

	public void setMsgResendTimes(int msgResendTimes) {
		MsgResendTimes = msgResendTimes;
	}
	
	/**
	 * Get the active thread count of thread pool
	 *
	 * @return count
	 */
	public int getThreadPoolSize()
	{
		return _threadPool.getPoolSize();
	}
	public int getThreadPoolActiveCount()
	{
		return _threadPool.getActiveCount();
	}
	/**
	 * Get thread pool task queue size now
	 *
	 * @return size
	 */
	public int getThreadPoolQueueSize()
	{
		return _threadPool.getQueue().size();
	}
}
